package com.lg.customRxJava;




public abstract class BasicFuseableObserver<T, R> implements Observer<T>, QueueDisposable<R> {

    protected final Observer<? super R> downstream;

    /** The upstream subscription. */
    protected Disposable upstream;

    /** The upstream's QueueDisposable if not null. */
    protected QueueDisposable<T> qd;

    /** Flag indicating no further onXXX event should be accepted. */
    protected boolean done;

    /** Holds the established fusion mode of the upstream. */
    protected int sourceMode;

    public BasicFuseableObserver(Observer<? super R> downstream) {
        this.downstream = downstream;
    }

    protected final int transitiveBoundaryFusion(int mode) {
        QueueDisposable<T> qd = this.qd;
        if (qd != null) {
            if ((mode & BOUNDARY) == 0) {
                int m = qd.requestFusion(mode);
                if (m != NONE) {
                    sourceMode = m;
                }
                return m;
            }
        }
        return NONE;
    }

    protected final void fail(Throwable t) {
      //  Exceptions.throwIfFatal(t);
        upstream.dispose();
        onError(t);
    }


    @Override
    public void dispose() {
        upstream.dispose();
    }

    @Override
    public boolean isDisposed() {
        return upstream.isDisposed();
    }

    @Override
    public boolean isEmpty() {
        return qd.isEmpty();
    }

    @Override
    public void clear() {
        qd.clear();
    }
    @Override
    public final boolean offer(R e) {
        throw new UnsupportedOperationException("Should not be called!");
    }

    @Override
    public final boolean offer(R v1, R v2) {
        throw new UnsupportedOperationException("Should not be called!");
    }


    @Override
    public void onError(Throwable t) {
        if (done) {
//            RxJavaPlugins.onError(t);
            return;
        }
        done = true;
        downstream.onError(t);
    }

    @Override
    public void onComplete() {
        if (done) {
            return;
        }
        done = true;
        downstream.onComplete();
    }

    @Override
    public final void onSubscribe(Disposable d) {
        if (DisposableHelper.validate(this.upstream, d)) {

            this.upstream = d;
            if (d instanceof QueueDisposable) {
                this.qd = (QueueDisposable<T>)d;
            }

            if (beforeDownstream()) {

                downstream.onSubscribe(this);

                afterDownstream();
            }

        }
    }

    protected boolean beforeDownstream() {
        return true;
    }

    /**
     * Override this to perform actions after the call to {@code actual.onSubscribe(this)} happened.
     */
    protected void afterDownstream() {
        // default no-op
    }
}
